import os
import time
import logging
import socket
from datetime import datetime, timedelta
from dotenv import load_dotenv
from tqdm import tqdm
import baostock as bs
import pymysql
from pymysql.cursors import DictCursor
import numpy as np

# ------------------------------ 全局配置 ------------------------------
load_dotenv()

# 日志配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('stock_k_combined_sync.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# 数据库配置
DB_CONFIG = {
    'host': os.getenv('DB_HOST', 'localhost'),
    'user': os.getenv('DB_USER', 'root'),
    'password': os.getenv('DB_PASSWORD', '123456'),
    'database': os.getenv('DB_NAME', 'stock_data'),
    'port': int(os.getenv('DB_PORT', 3306)),
    'charset': 'utf8mb4'
}

# 股票市场收盘时间
MARKET_CLOSE_HOUR = 15
MARKET_CLOSE_MINUTE = 0

# 核心字段配置（新增3个关键分析字段）
DAY_K_FIELDS = [
    'code', 'date', 'open', 'high', 'low', 'close',
    'volume', 'amount', 'adjustflag', 'turn', 'pctChg',
    'last_day_close'
]

BASE_K_FIELDS = [
    'code', 'date', 'open', 'high', 'low', 'close',
    'volume', 'amount', 'adjustflag', 'turn', 'pctChg'
]

# 周线新增字段：3个关键分析字段+中文前缀标准化
K_CONFIG = {
    'day': {
        'table': 'stock_day_k_history',
        'progress_file': 'processed_day_stocks.txt',
        'freq': 'd',
        'fields': DAY_K_FIELDS,
        'last_period_close_field': 'last_day_close',
        'api_interval': 0.2,
        'days_limit': 1000,
        'retry_count': 2
    },
    'week': {
        'table': 'policy_week_k',
        'progress_file': 'processed_week_stocks.txt',
        'freq': 'w',
        # 新增字段：突破60/120周线标记 + 均线粘合突破标记
        'fields': BASE_K_FIELDS + [
            'last_week_close', 'ma5', 'ma10', 'ma20', 'ma30', 'ma60', 'ma120',
            'analysis_result', 'trend_direction', 'ma5_trend',
            'distance_to_ma60_pct', 'distance_to_ma120_pct',
            'rs_cumulative_pct', 'ma_multiple_trend',
            'volume_price_fit', 'boom_prob_score',
            'is_break_ma60_week',    # 新增：是否突破60周线（开盘<60周线+收盘>60周线）
            'is_break_ma120_week',   # 新增：是否突破120周线（开盘<120周线+收盘>120周线）
            'is_ma_combined_break'   # 新增：5/10/20/30周线粘合且收盘站上
        ],
        'last_period_close_field': 'last_week_close',
        'api_interval': 0.3,
        'min_valid_weeks': 120,
        'recent_threshold_days': 90,
        'ma_periods': {
            'ma5': 5, 'ma10': 10, 'ma20': 20, 'ma30': 30, 'ma60': 60, 'ma120': 120
        },
        'break_threshold_pct': 0.02,
        'break_confirm_pct': 0.01,
        'trend_window': 12,
        'volume_window': 12,
        'rs_cumulative_window': 12,
        'boom_score_weights': {
            'trend_strength': 0.3, 'ma_multiple': 0.25, 'volume_price_fit': 0.2,
            'distance_to_ma': 0.15, 'volume_trend': 0.1
        },
        'ma_combined_threshold': 0.01  # 新增：均线粘合阈值（1%，即4条均线最大最小差值≤1%）
    },
    'month': {
        'table': 'stock_month_k_history',
        'progress_file': 'processed_month_stocks.txt',
        'freq': 'm',
        'fields': BASE_K_FIELDS + ['last_month_close'],
        'last_period_close_field': 'last_month_close',
        'api_interval': 0.5,
        'min_valid_months': 3,
        'recent_threshold_days': 180
    }
}

# 重试配置
STOCK_LIST_RETRY = 10
RETRY_DELAY_BASE = 2
FORCE_REPROCESS = False
START_DATE = '2010-01-01'


# ------------------------------ 公共工具函数 ------------------------------
def init_combined_database():
    """初始化数据库表结构（含新增3个字段）"""
    try:
        conn = pymysql.connect(
            host=DB_CONFIG['host'],
            user=DB_CONFIG['user'],
            password=DB_CONFIG['password'],
            port=DB_CONFIG['port'],
            charset=DB_CONFIG['charset']
        )
        cursor = conn.cursor()
        cursor.execute(f"CREATE DATABASE IF NOT EXISTS {DB_CONFIG['database']}")
        conn.select_db(DB_CONFIG['database'])

        for k_type in K_CONFIG.keys():
            cfg = K_CONFIG[k_type]
            if k_type == 'day':
                create_table_sql = f"""
                    CREATE TABLE IF NOT EXISTS {cfg['table']} (
                        id VARCHAR(50) NOT NULL COMMENT '唯一标识（股票代码_日期）',
                        code VARCHAR(20) NOT NULL COMMENT '股票代码（纯数字）',
                        date DATE NOT NULL COMMENT '日K线日期',
                        open DECIMAL(10, 3) NULL COMMENT '开盘价（保留3位小数）',
                        high DECIMAL(10, 3) NULL COMMENT '最高价（保留3位小数）',
                        low DECIMAL(10, 3) NULL COMMENT '最低价（保留3位小数）',
                        close DECIMAL(10, 3) NULL COMMENT '收盘价（保留3位小数）',
                        volume DECIMAL(12, 2) NULL COMMENT '成交量（万手）',
                        amount DECIMAL(12, 2) NULL COMMENT '成交额（亿元）',
                        adjustflag TINYINT NULL COMMENT '复权类型（2=后复权）',
                        turn DECIMAL(10, 2) NULL COMMENT '换手率（%）',
                        pctChg DECIMAL(10, 2) NULL COMMENT '涨跌幅（%）',
                        last_day_close DECIMAL(10, 3) NULL COMMENT '上一日收盘价',
                        insert_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
                        PRIMARY KEY (id),
                        INDEX idx_code_date (code, date),
                        INDEX idx_date (date)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
                """
            elif k_type == 'week':
                # 周线表：新增3个关键字段+中文前缀注释
                create_table_sql = f"""
                    CREATE TABLE IF NOT EXISTS {cfg['table']} (
                        id VARCHAR(50) NOT NULL COMMENT '唯一标识（股票代码_日期）',
                        code VARCHAR(20) NOT NULL COMMENT '股票代码（纯数字）',
                        date DATE NOT NULL COMMENT '周K线日期',
                        open DECIMAL(10, 3) NULL COMMENT '开盘价（保留3位小数）',
                        high DECIMAL(10, 3) NULL COMMENT '最高价（保留3位小数）',
                        low DECIMAL(10, 3) NULL COMMENT '最低价（保留3位小数）',
                        close DECIMAL(10, 3) NULL COMMENT '收盘价（保留3位小数）',
                        last_week_close DECIMAL(10, 3) NULL COMMENT '上周期收盘价',
                        volume DECIMAL(12, 2) NULL COMMENT '成交量（万手）',
                        amount DECIMAL(12, 2) NULL COMMENT '成交额（亿元）',
                        adjustflag TINYINT NULL COMMENT '复权类型（2=后复权）',
                        turn DECIMAL(10, 2) NULL COMMENT '换手率（%）',
                        pctChg DECIMAL(10, 2) NULL COMMENT '涨跌幅（%）',
                        ma5 VARCHAR(20) NULL COMMENT '5周线（保留3位小数）',
                        ma10 VARCHAR(20) NULL COMMENT '10周线（保留3位小数）',
                        ma20 VARCHAR(20) NULL COMMENT '20周线（保留3位小数）',
                        ma30 VARCHAR(20) NULL COMMENT '30周线（保留3位小数）',
                        ma60 VARCHAR(20) NULL COMMENT '60周线（保留3位小数）',
                        ma120 VARCHAR(20) NULL COMMENT '120周线（保留3位小数）',
                        analysis_result VARCHAR(100) NULL COMMENT '突破分析结果（前缀_描述格式）',
                        trend_direction VARCHAR(30) NULL COMMENT '趋势方向（向上_强/向上_弱/向下_强/向下_弱/横盘_无）',
                        ma5_trend VARCHAR(30) NULL COMMENT '5周线走势（向上_xxx/向下_xxx）',
                        distance_to_ma60_pct DECIMAL(10, 4) NULL COMMENT '距离60周线百分比（正=上方）',
                        distance_to_ma120_pct DECIMAL(10, 4) NULL COMMENT '距离120周线百分比（正=上方）',
                        rs_cumulative_pct DECIMAL(10, 2) NULL COMMENT '12周累计涨幅（%）',
                        ma_multiple_trend VARCHAR(30) NULL COMMENT '多均线排列（排列_多头/排列_空头/排列_震荡）',
                        volume_price_fit VARCHAR(20) NULL COMMENT '量价配合度（量价_良好/量价_一般/量价_背离）',
                        boom_prob_score DECIMAL(5, 1) NULL COMMENT '暴涨概率评分（0-100）',
                        is_break_ma60_week VARCHAR(10) NULL COMMENT '是否突破60周线（开盘<60周线且收盘>60周线：是/否）',
                        is_break_ma120_week VARCHAR(10) NULL COMMENT '是否突破120周线（开盘<120周线且收盘>120周线：是/否）',
                        is_ma_combined_break VARCHAR(10) NULL COMMENT '5/10/20/30周线是否粘合且收盘站上：是/否',
                        insert_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
                        PRIMARY KEY (id),
                        INDEX idx_code_date (code, date),
                        INDEX idx_date (date),
                        INDEX idx_boom_score (boom_prob_score),
                        INDEX idx_break_ma60 (is_break_ma60_week),
                        INDEX idx_break_ma120 (is_break_ma120_week),
                        INDEX idx_ma_combined (is_ma_combined_break)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
                """
            else:
                create_table_sql = f"""
                    CREATE TABLE IF NOT EXISTS {cfg['table']} (
                        id VARCHAR(50) NOT NULL COMMENT '唯一标识（股票代码_日期）',
                        code VARCHAR(20) NOT NULL COMMENT '股票代码（纯数字）',
                        date DATE NOT NULL COMMENT '{k_type}K线日期',
                        open DECIMAL(10, 3) NULL COMMENT '开盘价（保留3位小数）',
                        high DECIMAL(10, 3) NULL COMMENT '最高价（保留3位小数）',
                        low DECIMAL(10, 3) NULL COMMENT '最低价（保留3位小数）',
                        close DECIMAL(10, 3) NULL COMMENT '收盘价（保留3位小数）',
                        {cfg['last_period_close_field']} DECIMAL(10, 3) NULL COMMENT '上周期收盘价',
                        volume DECIMAL(12, 2) NULL COMMENT '成交量（万手）',
                        amount DECIMAL(12, 2) NULL COMMENT '成交额（亿元）',
                        adjustflag TINYINT NULL COMMENT '复权类型',
                        turn DECIMAL(10, 2) NULL COMMENT '换手率（%）',
                        pctChg DECIMAL(10, 2) NULL COMMENT '涨跌幅（%）',
                        insert_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
                        PRIMARY KEY (id),
                        INDEX idx_code_date (code, date),
                        INDEX idx_date (date)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
                """
            cursor.execute(create_table_sql)
            logger.info(f"{k_type.upper()}K表初始化成功（表名：{cfg['table']}）")

        conn.commit()
        logger.info("=== 日K+周K+月K数据库表结构初始化完成（含新增3个关键字段） ===")
    except Exception as e:
        logger.error(f"数据库初始化失败：{str(e)}", exc_info=True)
        raise
    finally:
        if 'conn' in locals():
            conn.close()


def get_recent_trading_date(force_prev_day=False):
    """获取最近的交易日"""
    now = datetime.now()
    is_before_close = (now.hour < MARKET_CLOSE_HOUR) or \
                      (now.hour == MARKET_CLOSE_HOUR and now.minute < MARKET_CLOSE_MINUTE)

    start_offset = 1 if (is_before_close or force_prev_day) else 0
    logger.info(f"当前时间：{now.strftime('%Y-%m-%d %H:%M:%S')}，{'未收盘' if is_before_close else '已收盘'}，"
                f"{'从昨天开始' if start_offset == 1 else '从今天开始'}检查交易日")

    for i in range(start_offset, 30):
        check_date = (now - timedelta(days=i)).strftime('%Y-%m-%d')
        try:
            lg = bs.login()
            if lg.error_code != '0':
                logger.warning(f"查询交易日[{check_date}]登录失败：{lg.error_msg}，将重试")
                bs.logout()
                continue

            rs = bs.query_trade_dates(start_date=check_date, end_date=check_date)
            if rs.error_code != '0':
                logger.warning(f"查询交易日[{check_date}]失败：{rs.error_msg}")
                bs.logout()
                continue

            if rs.next():
                is_trading = rs.get_row_data()[1] == '1'
                bs.logout()
                if is_trading:
                    logger.info(f"找到有效交易日：{check_date}")
                    return check_date

            bs.logout()
            logger.debug(f"{check_date}不是交易日，继续检查前一天")
        except socket.error as se:
            logger.warning(f"查询交易日[{check_date}]套接字错误：{str(se)}，将重试")
            time.sleep(1)
        except Exception as e:
            logger.warning(f"查询交易日[{check_date}]异常：{str(e)}，将重试")
            time.sleep(1)

    safe_date = "2024-12-31"
    logger.warning(f"无法获取有效交易日，使用安全日期：{safe_date}")
    return safe_date


def get_all_stock_codes():
    """获取所有A股代码"""
    trade_date = get_recent_trading_date()

    for retry in range(STOCK_LIST_RETRY):
        if retry > 0:
            trade_date = get_recent_trading_date(force_prev_day=True)
            logger.info(f"第{retry + 1}次重试，使用更早的交易日：{trade_date}")

        retry_delay = RETRY_DELAY_BASE ** min(retry, 5)
        try:
            lg = bs.login()
            if lg.error_code != '0':
                logger.error(f"股票列表获取失败（登录错误，重试{retry + 1}/{STOCK_LIST_RETRY}）：{lg.error_msg}")
                bs.logout()
                time.sleep(retry_delay)
                continue

            socket.setdefaulttimeout(15)
            rs = bs.query_all_stock(trade_date)
            if rs.error_code != '0':
                logger.error(f"股票列表获取失败（接口错误，重试{retry + 1}/{STOCK_LIST_RETRY}）：{rs.error_msg}")
                bs.logout()
                time.sleep(retry_delay)
                continue

            stock_codes = []
            while (rs.error_code == '0') and rs.next():
                code = rs.get_row_data()[0]
                if code.startswith(('sh.60', 'sh.688', 'sz.00', 'sz.30')):
                    stock_codes.append(code)

            if len(stock_codes) < 3000:
                logger.warning(f"第{retry + 1}次获取到股票数量不足（{len(stock_codes)}只），可能数据异常，将重试")
                bs.logout()
                time.sleep(retry_delay)
                continue

            bs.logout()
            logger.info(f"=== 成功获取{len(stock_codes)}只A股股票代码（重试{retry + 1}/{STOCK_LIST_RETRY}） ===")
            return stock_codes

        except socket.timeout:
            logger.warning(f"股票列表获取超时（重试{retry + 1}/{STOCK_LIST_RETRY}），将重试")
            bs.logout()
            time.sleep(retry_delay)
        except socket.error as se:
            logger.warning(f"股票列表获取套接字错误（重试{retry + 1}/{STOCK_LIST_RETRY}）：{str(se)}，将重试")
            bs.logout()
            time.sleep(retry_delay)
        except Exception as e:
            logger.error(f"股票列表获取异常（重试{retry + 1}/{STOCK_LIST_RETRY}）：{str(e)}", exc_info=True)
            bs.logout()
            time.sleep(retry_delay)

    logger.error("=" * 50)
    logger.error("严重错误：所有尝试均未能从Baostock获取有效股票列表")
    logger.error("可能原因：")
    logger.error("1. Baostock服务临时故障（请稍后再试）")
    logger.error("2. 网络连接不稳定或被限制访问")
    logger.error("3. 近期无有效交易日（如长假期间）")
    logger.error("=" * 50)
    raise RuntimeError("无法获取股票列表，任务终止")


def load_valid_processed_stocks(k_type, all_stocks):
    """加载已处理股票"""
    cfg = K_CONFIG[k_type]
    if not os.path.exists(cfg['progress_file']):
        logger.info(f"{k_type.upper()}K进度文件不存在，将处理所有股票")
        return []

    with open(cfg['progress_file'], 'r', encoding='utf-8') as f:
        progress_codes = [line.strip() for line in f if line.strip()]

    valid_progress = [code for code in progress_codes if code in all_stocks]
    if not valid_progress:
        logger.info(f"{k_type.upper()}K进度文件中无有效股票代码，将处理所有股票")
        return []

    try:
        conn = pymysql.connect(**DB_CONFIG)
        cursor = conn.cursor()
        clean_codes = [code.split('.')[1] for code in valid_progress]
        placeholders = ', '.join(['%s'] * len(clean_codes))
        cursor.execute(f"""
            SELECT DISTINCT code FROM {cfg['table']} 
            WHERE code IN ({placeholders})
        """, tuple(clean_codes))
        db_codes = [row[0] for row in cursor.fetchall()]
        db_valid_codes = [code for code in valid_progress
                          if clean_stock_code(code) in db_codes]

        logger.info(f"{k_type.upper()}K验证后，已处理股票：{len(db_valid_codes)}只")
        return db_valid_codes
    except Exception as e:
        logger.warning(f"{k_type.upper()}K验证已处理股票失败，将重新处理：{str(e)}")
        return []
    finally:
        if 'conn' in locals():
            conn.close()


def save_processed_stock(k_type, code):
    """保存已处理股票"""
    cfg = K_CONFIG[k_type]
    with open(cfg['progress_file'], 'a', encoding='utf-8') as f:
        f.write(f"{code}\n")


def clean_stock_code(raw_code):
    """清洗股票代码"""
    return raw_code.split('.')[1] if '.' in raw_code else raw_code


def generate_id(code, date):
    """生成唯一ID"""
    date_str = date.replace('-', '')
    return f"{code}_{date_str}"


def convert_volume(volume):
    """转换成交量单位（股→万手）"""
    if volume is None:
        return None
    try:
        return round(volume / 1_000_000, 2)
    except (TypeError, ValueError):
        return None


def convert_amount(amount):
    """转换成交额单位（元→亿元）"""
    if amount is None:
        return None
    try:
        return round(amount / 100_000_000, 2)
    except (TypeError, ValueError):
        return None


def calculate_last_period_close(k_type, k_data):
    """计算上周期收盘价（保留3位小数）"""
    if not k_data:
        return []

    sorted_data = sorted(k_data, key=lambda x: x['date'])
    cfg = K_CONFIG[k_type]
    last_close_field = cfg['last_period_close_field']

    for i in range(1, len(sorted_data)):
        prev_date = datetime.strptime(sorted_data[i - 1]['date'], '%Y-%m-%d')
        curr_date = datetime.strptime(sorted_data[i]['date'], '%Y-%m-%d')
        if curr_date > prev_date:
            sorted_data[i][last_close_field] = round(sorted_data[i - 1]['close'], 3) if sorted_data[i - 1][
                                                                                            'close'] is not None else None
            if sorted_data[i][last_close_field] is not None and sorted_data[i][last_close_field] != 0 and \
                    sorted_data[i]['close'] is not None:
                sorted_data[i]['pctChg'] = round(
                    (sorted_data[i]['close'] - sorted_data[i][last_close_field]) /
                    sorted_data[i][last_close_field] * 100, 2
                )
        else:
            logger.warning(
                f"{k_type.upper()}K数据乱序：{sorted_data[i]['date']} ≤ {sorted_data[i - 1]['date']}，跳过计算")
            sorted_data[i][last_close_field] = None

    sorted_data[0][last_close_field] = None
    sorted_data[0]['pctChg'] = None
    return sorted_data


# ------------------------------ 辅助算法：专业趋势分析工具 ------------------------------
def calculate_trend_strength(prices):
    """专业趋势强度计算（基于线性回归+波动率）"""
    if len(prices) < 5:
        return 0

    x = np.arange(len(prices))
    y = np.array(prices)
    slope, intercept = np.polyfit(x, y, 1)
    volatility = np.std(y)
    if volatility == 0:
        return 0

    trend_strength = min(abs(slope) / volatility * 100, 100)
    return round(trend_strength, 1)


def judge_ma_multiple_trend(ma_values):
    """判断多均线排列趋势（返回前缀_描述格式）"""
    if any(val is None or val == '数据缺失' for val in ma_values):
        return '排列_震荡'

    ma5, ma10, ma20, ma30 = [float(val) for val in ma_values]
    long_condition = (ma5 > ma10 * 1.005) and (ma10 > ma20 * 1.005) and (ma20 > ma30 * 1.005)
    short_condition = (ma5 < ma10 * 0.995) and (ma10 < ma20 * 0.995) and (ma20 < ma30 * 0.995)

    if long_condition:
        return '排列_多头'
    elif short_condition:
        return '排列_空头'
    else:
        return '排列_震荡'


def calculate_volume_price_fit(weeks_data, window=12):
    """计算量价配合度（返回前缀_描述格式）"""
    if len(weeks_data) < window:
        return '量价_一般'

    up_weeks_vol = []
    down_weeks_vol = []

    for week in weeks_data:
        pct_chg = week.get('pctChg')
        volume = week.get('volume')
        if pct_chg is None or volume is None or volume == 0:
            continue

        if pct_chg > 0:
            up_weeks_vol.append(volume)
        elif pct_chg < 0:
            down_weeks_vol.append(volume)

    if len(up_weeks_vol) == 0 or len(down_weeks_vol) == 0:
        return '量价_一般'

    avg_up_vol = sum(up_weeks_vol) / len(up_weeks_vol)
    avg_down_vol = sum(down_weeks_vol) / len(down_weeks_vol)
    fit_ratio = avg_up_vol / avg_down_vol

    if fit_ratio > 1.2:
        return '量价_良好'
    elif fit_ratio < 0.8:
        return '量价_背离'
    else:
        return '量价_一般'


def judge_ma_combined(ma5, ma10, ma20, ma30, threshold=0.01):
    """判断5/10/20/30周线是否粘合（阈值默认1%）"""
    if None in [ma5, ma10, ma20, ma30] or 0 in [ma5, ma10, ma20, ma30]:
        return False

    ma_list = [ma5, ma10, ma20, ma30]
    max_ma = max(ma_list)
    min_ma = min(ma_list)
    # 粘合判断：(最大值-最小值)/最小值 ≤ 阈值
    return (max_ma - min_ma) / min_ma <= threshold


# ------------------------------ 核心：周线多维度分析（含新增字段+中文前缀） ------------------------------
def calculate_week_ma_and_break_analysis(week_k_data):
    """
    周线多维度分析（优化中文前缀+新增3个关键字段）：
    1. 中文描述统一为“前缀_描述”格式（便于搜索）
    2. 新增is_break_ma60_week：开盘<60周线+收盘>60周线（是/否）
    3. 新增is_break_ma120_week：开盘<120周线+收盘>120周线（是/否）
    4. 新增is_ma_combined_break：5/10/20/30周线粘合且收盘站上（是/否）
    """
    if not week_k_data:
        return []

    cfg = K_CONFIG['week']
    ma_periods = cfg['ma_periods']
    break_threshold_pct = cfg['break_threshold_pct']
    break_confirm_pct = cfg['break_confirm_pct']
    trend_window = cfg['trend_window']
    volume_window = cfg['volume_window']
    rs_window = cfg['rs_cumulative_window']
    boom_weights = cfg['boom_score_weights']
    ma_combined_threshold = cfg['ma_combined_threshold']  # 均线粘合阈值

    sorted_week_data = sorted(week_k_data, key=lambda x: x['date'])
    total_weeks = len(sorted_week_data)

    for i in range(total_weeks):
        current = sorted_week_data[i]
        current_close = current.get('close')
        current_open = current.get('open')  # 用于突破判断
        current_date = current['date']
        code = current['code']

        # 初始化所有字段（含新增3个字段，默认“否”）
        for ma_field in ma_periods.keys():
            current[ma_field] = '数据缺失'
        current['analysis_result'] = '突破_数据不足'
        current['trend_direction'] = '横盘_无'  # 前缀_描述
        current['ma5_trend'] = '向下_未在5周线'  # 前缀_描述
        current['distance_to_ma60_pct'] = None
        current['distance_to_ma120_pct'] = None
        current['rs_cumulative_pct'] = None
        current['ma_multiple_trend'] = '排列_震荡'  # 前缀_描述
        current['volume_price_fit'] = '量价_一般'  # 前缀_描述
        current['boom_prob_score'] = 0.0
        current['is_break_ma60_week'] = '否'  # 新增字段默认
        current['is_break_ma120_week'] = '否'  # 新增字段默认
        current['is_ma_combined_break'] = '否'  # 新增字段默认

        # 跳过关键数据为空的情况
        if current_close is None or current_open is None:
            logger.debug(f"周线[{code}_{current_date}]开盘价/收盘价为空，跳过分析")
            continue

        # 1. 计算所有均线
        ma_values = {}
        for ma_field, period in ma_periods.items():
            if i + 1 >= period:
                ma_window = sorted_week_data[i - period + 1: i + 1]
                ma_closes = [w['close'] for w in ma_window if w['close'] is not None]
                if len(ma_closes) == period:
                    ma_value = round(sum(ma_closes) / period, 3)
                    current[ma_field] = f"{ma_value:.3f}"
                    ma_values[ma_field] = ma_value
                else:
                    current[ma_field] = '数据缺失'
            else:
                current[ma_field] = '数据不足'

        # 2. 趋势方向（前缀_描述格式）
        if i >= trend_window - 1:
            trend_prices = [sorted_week_data[j]['close'] for j in range(i - trend_window + 1, i + 1)
                            if sorted_week_data[j]['close'] is not None]
            if len(trend_prices) >= trend_window * 0.8:
                trend_strength = calculate_trend_strength(trend_prices)
                slope = (trend_prices[-1] - trend_prices[0]) / (len(trend_prices) - 1)

                if slope > 0:
                    current['trend_direction'] = '向上_强' if trend_strength >= 60 else '向上_弱'
                elif slope < 0:
                    current['trend_direction'] = '向下_强' if trend_strength >= 60 else '向下_弱'
                else:
                    current['trend_direction'] = '横盘_无'

        # 3. 5周线走势（前缀_描述格式）
        current_ma5 = ma_values.get('ma5')
        if current_ma5 is not None:
            if current_close > current_ma5:
                consecutive_weeks = 0
                for j in range(max(0, i - 4), i + 1):
                    week_j = sorted_week_data[j]
                    week_ma5 = week_j.get('ma5')
                    week_close = week_j.get('close')
                    if week_ma5 not in ('数据缺失', '数据不足') and week_close is not None:
                        if week_close > float(week_ma5):
                            consecutive_weeks += 1
                if consecutive_weeks >= 5:
                    current['ma5_trend'] = '向上_沿5周线（≥5周）'
                else:
                    current['ma5_trend'] = f'向上_在5周线（{consecutive_weeks}周）'
            else:
                current['ma5_trend'] = '向下_未在5周线'

        # 4. 距离60周线/120周线百分比
        for ma_field, pct_field in [('ma60', 'distance_to_ma60_pct'), ('ma120', 'distance_to_ma120_pct')]:
            ma_val = ma_values.get(ma_field)
            if ma_val is not None and ma_val != 0:
                pct = (current_close - ma_val) / ma_val * 100
                current[pct_field] = round(pct, 4)

        # 5. 12周累计涨幅
        if i >= rs_window - 1:
            start_close = sorted_week_data[i - rs_window + 1]['close']
            if start_close is not None and start_close != 0:
                cumulative_pct = (current_close - start_close) / start_close * 100
                current['rs_cumulative_pct'] = round(cumulative_pct, 2)

        # 6. 多均线排列（前缀_描述格式）
        multiple_ma_list = [ma_values.get('ma5'), ma_values.get('ma10'), ma_values.get('ma20'), ma_values.get('ma30')]
        if all(val is not None for val in multiple_ma_list):
            current['ma_multiple_trend'] = judge_ma_multiple_trend([
                f"{val:.3f}" for val in multiple_ma_list
            ])

        # 7. 量价配合度（前缀_描述格式）
        if i >= volume_window - 1:
            recent_weeks = sorted_week_data[i - volume_window + 1: i + 1]
            current['volume_price_fit'] = calculate_volume_price_fit(recent_weeks, volume_window)

        # 8. 突破半年线分析（前缀_描述格式）
        ma120_val = ma_values.get('ma120')
        if ma120_val is not None:
            price_gap_pct = (current_close - ma120_val) / ma120_val if ma120_val != 0 else 0

            if i >= 1:
                last_week = sorted_week_data[i - 1]
                last_close = last_week.get('close')
                last_ma120 = last_week.get('ma120')
                if last_close is not None and last_ma120 not in ('数据缺失', '数据不足'):
                    last_ma120_val = float(last_ma120)
                    if (last_close < last_ma120_val) and (current_close > ma120_val):
                        if price_gap_pct >= break_confirm_pct:
                            current['analysis_result'] = '突破_刚突破半年线'
                            logger.info(f"股票[{code}]在[{current_date}]刚突破半年线（MA120={ma120_val:.3f}）")
                        else:
                            current['analysis_result'] = '突破_不明显（涨幅<1%）'

            if current['analysis_result'] == '突破_数据不足':
                if current_close < ma120_val:
                    current['analysis_result'] = '突破_待突破半年线' if abs(
                        price_gap_pct) <= break_threshold_pct else '突破_远离半年线'
                else:
                    if i >= 1:
                        last_week = sorted_week_data[i - 1]
                        last_close = last_week.get('close')
                        last_ma120 = last_week.get('ma120')
                        if last_close is not None and last_ma120 not in ('数据缺失', '数据不足'):
                            current['analysis_result'] = '突破_已站上（≥2周）' if last_close > float(
                                last_ma120) else '突破_刚站上（1周）'
                    else:
                        current['analysis_result'] = '突破_刚站上（1周）'

        # 9. 暴涨概率评分
        boom_score = 0.0
        trend_score = 100 if current['trend_direction'] == '向上_强' else 60 if current['trend_direction'] == '向上_弱' else 0
        boom_score += trend_score * boom_weights['trend_strength']

        ma_score = 100 if current['ma_multiple_trend'] == '排列_多头' else 0
        boom_score += ma_score * boom_weights['ma_multiple']

        vp_score = 100 if current['volume_price_fit'] == '量价_良好' else 50 if current['volume_price_fit'] == '量价_一般' else 0
        boom_score += vp_score * boom_weights['volume_price_fit']

        distance_score = 0
        if current['distance_to_ma60_pct'] is not None:
            distance = current['distance_to_ma60_pct']
            if 0 < distance <= 5:
                distance_score = 100
            elif -2 <= distance <= 0:
                distance_score = 80
            elif 5 < distance <= 10:
                distance_score = 60
        boom_score += distance_score * boom_weights['distance_to_ma']

        volume_score = 0
        if i >= 2 * volume_window - 1:
            recent_volume = [sorted_week_data[j]['volume'] for j in range(i - volume_window + 1, i + 1)
                             if sorted_week_data[j]['volume'] is not None]
            history_volume = [sorted_week_data[j]['volume'] for j in
                              range(i - 2 * volume_window + 1, i - volume_window + 1)
                              if sorted_week_data[j]['volume'] is not None]
            if len(recent_volume) >= volume_window * 0.8 and len(history_volume) >= volume_window * 0.8:
                avg_recent = sum(recent_volume) / len(recent_volume)
                avg_history = sum(history_volume) / len(history_volume)
                if avg_history != 0 and (avg_recent / avg_history) > 1.2:
                    volume_score = 100
        boom_score += volume_score * boom_weights['volume_trend']

        current['boom_prob_score'] = round(min(boom_score, 100.0), 1)

        # 10. 新增字段1：是否突破60周线（开盘<60周线+收盘>60周线）
        ma60_val = ma_values.get('ma60')
        if ma60_val is not None and ma60_val != 0:
            if current_open < ma60_val and current_close > ma60_val:
                current['is_break_ma60_week'] = '是'

        # 11. 新增字段2：是否突破120周线（开盘<120周线+收盘>120周线）
        ma120_val = ma_values.get('ma120')
        if ma120_val is not None and ma120_val != 0:
            if current_open < ma120_val and current_close > ma120_val:
                current['is_break_ma120_week'] = '是'

        # 12. 新增字段3：5/10/20/30周线粘合且收盘站上
        ma5_val = ma_values.get('ma5')
        ma10_val = ma_values.get('ma10')
        ma20_val = ma_values.get('ma20')
        ma30_val = ma_values.get('ma30')
        if all([ma5_val, ma10_val, ma20_val, ma30_val]):  # 四条均线均有效
            is_combined = judge_ma_combined(ma5_val, ma10_val, ma20_val, ma30_val, ma_combined_threshold)
            if is_combined and current_close > max(ma5_val, ma10_val, ma20_val, ma30_val):
                current['is_ma_combined_break'] = '是'
                logger.info(f"股票[{code}]在[{current_date}]满足均线粘合突破：5/10/20/30周线粘合，收盘价站上")

        # 测试日志：输出新增字段结果
        if code == '600004':
            logger.debug(f"600004周线[{current_date}]："
                         f"突破60周线={current['is_break_ma60_week']}，"
                         f"突破120周线={current['is_break_ma120_week']}，"
                         f"均线粘合突破={current['is_ma_combined_break']}，"
                         f"趋势方向={current['trend_direction']}，"
                         f"量价配合={current['volume_price_fit']}")

    return sorted_week_data


# ------------------------------ K线计算逻辑 ------------------------------
def calculate_week_k_from_day_k(day_k_data):
    """用日线数据计算周线（按自然周划分，确保后复权）"""
    if not day_k_data:
        return []

    week_groups = {}
    for day in day_k_data:
        date = datetime.strptime(day['date'], '%Y-%m-%d')
        year, week_num, _ = date.isocalendar()
        key = f"{year}-W{week_num:02d}"
        if key not in week_groups:
            week_groups[key] = []
        week_groups[key].append(day)

    week_k_list = []
    for key, days in week_groups.items():
        sorted_days = sorted(days, key=lambda x: x['date'])
        week_end_date = sorted_days[-1]['date']
        code = days[0]['code']

        week_k = {
            'id': generate_id(code, week_end_date),
            'code': code,
            'date': week_end_date,
            'open': round(sorted_days[0]['open'], 3) if sorted_days[0]['open'] is not None else None,
            'high': round(max(day['high'] for day in sorted_days if day['high'] is not None), 3) if any(
                day['high'] is not None for day in sorted_days) else None,
            'low': round(min(day['low'] for day in sorted_days if day['low'] is not None), 3) if any(
                day['low'] is not None for day in sorted_days) else None,
            'close': round(sorted_days[-1]['close'], 3) if sorted_days[-1]['close'] is not None else None,
            'volume': sum(day['volume'] for day in sorted_days if day['volume'] is not None),
            'amount': sum(day['amount'] for day in sorted_days if day['amount'] is not None),
            'adjustflag': 2,
            'turn': round(sum(day['turn'] for day in sorted_days if day['turn'] is not None), 2) if any(
                day['turn'] is not None for day in sorted_days) else None,
            'pctChg': None
        }
        week_k_list.append(week_k)

    week_k_with_last_close = calculate_last_period_close('week', week_k_list)
    week_k_with_ma_analysis = calculate_week_ma_and_break_analysis(week_k_with_last_close)
    return week_k_with_ma_analysis


def calculate_month_k_from_day_k(day_k_data):
    """用日线数据计算月线（按自然月划分）"""
    if not day_k_data:
        return []

    month_groups = {}
    for day in day_k_data:
        date = datetime.strptime(day['date'], '%Y-%m-%d')
        key = f"{date.year}-{date.month:02d}"
        if key not in month_groups:
            month_groups[key] = []
        month_groups[key].append(day)

    month_k_list = []
    for key, days in month_groups.items():
        sorted_days = sorted(days, key=lambda x: x['date'])
        month_end_date = sorted_days[-1]['date']
        code = days[0]['code']

        month_k = {
            'id': generate_id(code, month_end_date),
            'code': code,
            'date': month_end_date,
            'open': round(sorted_days[0]['open'], 3) if sorted_days[0]['open'] is not None else None,
            'high': round(max(day['high'] for day in sorted_days if day['high'] is not None), 3) if any(
                day['high'] is not None for day in sorted_days) else None,
            'low': round(min(day['low'] for day in sorted_days if day['low'] is not None), 3) if any(
                day['low'] is not None for day in sorted_days) else None,
            'close': round(sorted_days[-1]['close'], 3) if sorted_days[-1]['close'] is not None else None,
            'volume': sum(day['volume'] for day in sorted_days if day['volume'] is not None),
            'amount': sum(day['amount'] for day in sorted_days if day['amount'] is not None),
            'adjustflag': 2,
            'turn': round(sum(day['turn'] for day in sorted_days if day['turn'] is not None), 2) if any(
                day['turn'] is not None for day in sorted_days) else None,
            'pctChg': None
        }
        month_k_list.append(month_k)

    return calculate_last_period_close('month', month_k_list)


# ------------------------------ 核心同步函数 ------------------------------
def _fetch_week_k_from_api(raw_code, cfg, global_login):
    """从接口获取周线数据（确保后复权）"""
    try:
        start_date = START_DATE
        fields_str = ",".join(BASE_K_FIELDS)

        rs = bs.query_history_k_data_plus(
            raw_code,
            fields_str,
            start_date=start_date,
            end_date=datetime.now().strftime('%Y-%m-%d'),
            frequency=cfg['freq'],
            adjustflag='2'
        )

        if rs.error_code != '0':
            logger.warning(f"WEEKK：股票{raw_code}接口错误：{rs.error_msg}")
            return []

        data_list = []
        clean_code = clean_stock_code(raw_code)
        while (rs.error_code == '0') and rs.next():
            row = rs.get_row_data()
            raw_data = dict(zip(BASE_K_FIELDS, row))

            processed_data = {
                'id': generate_id(clean_code, raw_data['date']),
                'code': clean_code,
                'date': raw_data['date']
            }

            for key in ['open', 'high', 'low', 'close']:
                if raw_data.get(key) and raw_data[key] != '':
                    processed_data[key] = round(float(raw_data[key]), 3)
                else:
                    processed_data[key] = None

            processed_data['turn'] = round(float(raw_data['turn']), 2) if raw_data.get('turn') and raw_data[
                'turn'] != '' else None
            processed_data['pctChg'] = round(float(raw_data['pctChg']), 2) if raw_data.get('pctChg') and raw_data[
                'pctChg'] != '' else None

            processed_data['volume'] = convert_volume(
                int(raw_data['volume']) if raw_data.get('volume') and raw_data['volume'] != '' else None
            )
            processed_data['amount'] = convert_amount(
                float(raw_data['amount']) if raw_data.get('amount') and raw_data['amount'] != '' else None
            )
            processed_data['adjustflag'] = 2

            data_list.append(processed_data)

        data_with_last_close = calculate_last_period_close('week', data_list)
        data_with_ma_analysis = calculate_week_ma_and_break_analysis(data_with_last_close)
        return data_with_ma_analysis
    except Exception as e:
        logger.error(f"WEEKK：股票{raw_code}接口数据解析失败：{str(e)}", exc_info=True)
        return []


def _fetch_month_k_from_api(raw_code, cfg, global_login):
    """从接口获取月线数据"""
    try:
        start_date = START_DATE
        fields_str = ",".join(BASE_K_FIELDS)

        rs = bs.query_history_k_data_plus(
            raw_code,
            fields_str,
            start_date=start_date,
            end_date=datetime.now().strftime('%Y-%m-%d'),
            frequency=cfg['freq'],
            adjustflag='2'
        )

        if rs.error_code != '0':
            logger.warning(f"MONTMK：股票{raw_code}接口错误：{rs.error_msg}")
            return []

        data_list = []
        clean_code = clean_stock_code(raw_code)
        while (rs.error_code == '0') and rs.next():
            row = rs.get_row_data()
            raw_data = dict(zip(BASE_K_FIELDS, row))

            processed_data = {
                'id': generate_id(clean_code, raw_data['date']),
                'code': clean_code,
                'date': raw_data['date']
            }

            for key in ['open', 'high', 'low', 'close']:
                if raw_data.get(key) and raw_data[key] != '':
                    processed_data[key] = round(float(raw_data[key]), 3)
                else:
                    processed_data[key] = None

            processed_data['turn'] = round(float(raw_data['turn']), 2) if raw_data.get('turn') and raw_data[
                'turn'] != '' else None
            processed_data['pctChg'] = round(float(raw_data['pctChg']), 2) if raw_data.get('pctChg') and raw_data[
                'pctChg'] != '' else None

            processed_data['volume'] = convert_volume(
                int(raw_data['volume']) if raw_data.get('volume') and raw_data['volume'] != '' else None
            )
            processed_data['amount'] = convert_amount(
                float(raw_data['amount']) if raw_data.get('amount') and raw_data['amount'] != '' else None
            )
            processed_data['adjustflag'] = 2

            data_list.append(processed_data)

        return calculate_last_period_close('month', data_list)
    except Exception as e:
        logger.error(f"MONTMK：股票{raw_code}接口数据解析失败：{str(e)}", exc_info=True)
        return []


def _has_recent_data(k_data, threshold_days):
    """检查是否包含近期数据"""
    if not k_data:
        return False
    latest_date = datetime.strptime(k_data[-1]['date'], '%Y-%m-%d')
    return (datetime.now() - latest_date) <= timedelta(days=threshold_days)


def fetch_stock_k_data(k_type, raw_code, global_login=False):
    """获取单只股票K线数据"""
    cfg = K_CONFIG[k_type]
    try:
        if not global_login:
            lg = bs.login()
            if lg.error_code != '0':
                logger.warning(f"{k_type.upper()}K：股票{raw_code}登录失败，跳过：{lg.error_msg}")
                return []

        if k_type == 'week':
            week_data = _fetch_week_k_from_api(raw_code, cfg, global_login)
            if len(week_data) >= cfg['min_valid_weeks'] and _has_recent_data(week_data, cfg['recent_threshold_days']):
                logger.info(f"WEEKK：股票{raw_code}使用接口数据（{len(week_data)}条，含新增字段分析）")
                return week_data
            logger.warning(f"WEEKK：股票{raw_code}接口数据不足（{len(week_data)}条），改用日线计算")
            day_data = fetch_stock_k_data('day', raw_code, global_login)
            return calculate_week_k_from_day_k(day_data)

        if k_type == 'month':
            month_data = _fetch_month_k_from_api(raw_code, cfg, global_login)
            if len(month_data) >= cfg['min_valid_months'] and _has_recent_data(month_data,
                                                                               cfg['recent_threshold_days']):
                logger.info(f"MONTMK：股票{raw_code}使用接口数据（{len(month_data)}条）")
                return month_data
            logger.warning(f"MONTMK：股票{raw_code}接口数据不足（{len(month_data)}条），改用日线计算")
            day_data = fetch_stock_k_data('day', raw_code, global_login)
            return calculate_month_k_from_day_k(day_data)

        if k_type == 'day':
            start_date = (datetime.now() - timedelta(days=cfg['days_limit'])).strftime('%Y-%m-%d')
            fields_str = ",".join([f for f in cfg['fields'] if f != cfg['last_period_close_field']])

            rs = bs.query_history_k_data_plus(
                raw_code,
                fields_str,
                start_date=start_date,
                end_date=datetime.now().strftime('%Y-%m-%d'),
                frequency=cfg['freq'],
                adjustflag='2'
            )

            if rs.error_code != '0':
                logger.warning(f"DAYK：股票{raw_code}数据获取失败：{rs.error_msg}")
                return []

            data_list = []
            clean_code = clean_stock_code(raw_code)
            while (rs.error_code == '0') and rs.next():
                row = rs.get_row_data()
                raw_fields = [f for f in cfg['fields'] if f != cfg['last_period_close_field']]
                raw_data = dict(zip(raw_fields, row))

                processed_data = {
                    'id': generate_id(clean_code, raw_data['date']),
                    'code': clean_code,
                    'date': raw_data['date']
                }

                for key in ['open', 'high', 'low', 'close']:
                    if raw_data.get(key) and raw_data[key] != '':
                        processed_data[key] = round(float(raw_data[key]), 3)
                    else:
                        processed_data[key] = None

                processed_data['turn'] = round(float(raw_data['turn']), 2) if raw_data.get('turn') and raw_data[
                    'turn'] != '' else None
                processed_data['pctChg'] = round(float(raw_data['pctChg']), 2) if raw_data.get('pctChg') and raw_data[
                    'pctChg'] != '' else None

                processed_data['volume'] = convert_volume(
                    int(raw_data['volume']) if raw_data.get('volume') and raw_data['volume'] != '' else None
                )
                processed_data['amount'] = convert_amount(
                    float(raw_data['amount']) if raw_data.get('amount') and raw_data['amount'] != '' else None
                )
                processed_data['adjustflag'] = 2

                data_list.append(processed_data)

            return calculate_last_period_close('day', data_list)

    except Exception as e:
        logger.error(f"{k_type.upper()}K：股票{raw_code}处理异常，跳过：{str(e)}", exc_info=True)
        return []
    finally:
        if not global_login:
            bs.logout()


def batch_insert_to_db(k_type, data_list):
    """批量插入/更新数据库（含新增3个字段）"""
    if not data_list:
        return 0

    cfg = K_CONFIG[k_type]
    try:
        conn = pymysql.connect(**DB_CONFIG, cursorclass=DictCursor)
        cursor = conn.cursor()
        insert_fields = ['id', 'code', 'date'] + [f for f in cfg['fields'] if f not in ['id', 'code', 'date']]

        update_fields = [f"{field}=VALUES({field})" for field in insert_fields if field != 'id']
        sql = f"""
            INSERT INTO {cfg['table']} ({','.join(insert_fields)})
            VALUES ({','.join(['%s'] * len(insert_fields))})
            ON DUPLICATE KEY UPDATE {','.join(update_fields)}
        """

        batch_size = 2000
        total_affected = 0
        for i in range(0, len(data_list), batch_size):
            batch = data_list[i:i + batch_size]
            values = [tuple(item[field] for field in insert_fields) for item in batch]
            affected = cursor.executemany(sql, values)
            total_affected += affected
            conn.commit()

        logger.info(f"{k_type.upper()}K：批量操作完成，影响{total_affected}条数据（新增+更新）")
        return total_affected
    except Exception as e:
        conn.rollback()
        logger.error(f"{k_type.upper()}K：数据库操作失败：{str(e)}", exc_info=True)
        return 0
    finally:
        if 'conn' in locals():
            conn.close()


def sync_single_k_type(k_type, all_stocks, global_login_success):
    """单类型K线同步"""
    cfg = K_CONFIG[k_type]
    logger.info(f"\n===== 开始{cfg['freq'].upper()}K线同步任务 =====")
    start_time = time.time()

    if FORCE_REPROCESS:
        need_process = all_stocks
    else:
        processed_stocks = load_valid_processed_stocks(k_type, all_stocks)
        need_process = [code for code in all_stocks if code not in processed_stocks]

    logger.info(f"{k_type.upper()}K：股票总数：{len(all_stocks)}只")
    logger.info(f"{k_type.upper()}K：需处理股票：{len(need_process)}只")
    if not need_process and not FORCE_REPROCESS:
        logger.info(f"{k_type.upper()}K：所有股票均已同步完成，跳过任务")
        return 0, 0

    total_success = 0
    total_data = 0

    for raw_code in tqdm(need_process, desc=f"{k_type.upper()}K处理进度"):
        time.sleep(cfg['api_interval'])
        clean_code = clean_stock_code(raw_code)

        k_data = None
        for retry in range(cfg.get('retry_count', 1) + 1):
            k_data = fetch_stock_k_data(k_type, raw_code, global_login_success)
            if len(k_data) > 0 or retry >= cfg.get('retry_count', 1):
                break
            logger.warning(f"{k_type.upper()}K：股票{raw_code}第{retry + 1}次重试（当前数据量：0）")
            time.sleep(1)

        if not k_data:
            logger.warning(
                f"{k_type.upper()}K：股票{raw_code}经{cfg.get('retry_count', 1) + 1}次尝试仍未获取到数据，跳过")
            continue

        affected = batch_insert_to_db(k_type, k_data)
        if affected <= 0:
            logger.warning(f"{k_type.upper()}K：股票{raw_code}首次插入失败，重试1次")
            time.sleep(1)
            affected = batch_insert_to_db(k_type, k_data)
            if affected <= 0:
                logger.error(f"{k_type.upper()}K：股票{raw_code}多次插入失败，放弃")
                continue

        save_processed_stock(k_type, raw_code)
        total_success += 1
        total_data += len(k_data)
        logger.info(f"{k_type.upper()}K：股票{raw_code}处理完成（累计：{total_success}/{len(need_process)}）")

    end_time = time.time()
    elapsed = int(end_time - start_time)
    logger.info(f"""
    ===== {k_type.upper()}K线同步任务完成 =====
    总处理股票数：{len(need_process)}只
    成功同步股票数：{total_success}只
    总插入数据量：{total_data}条
    耗时：{elapsed // 3600}小时{(elapsed % 3600) // 60}分钟{elapsed % 60}秒
    """)
    return total_success, total_data


# ------------------------------ 主函数 ------------------------------
def main():
    logger.info("=" * 50)
    logger.info("===== 开始日K→周K→月K连续同步任务（含新增字段+中文前缀优化） =====")
    logger.info("=" * 50)
    overall_start = time.time()

    init_combined_database()
    all_stocks = get_all_stock_codes()
    if not all_stocks:
        logger.error("未获取到任何股票代码，任务终止")
        return

    # 全局登录
    global_login_success = False
    try:
        lg = bs.login()
        if lg.error_code == '0':
            global_login_success = True
            logger.info("\n=== Baostock全局登录成功 ===")
        else:
            logger.warning(f"\n=== Baostock全局登录失败，将使用单股票登录模式：{lg.error_msg} ===")

        # 依次同步日K→周K→月K（按需注释）
        # day_success, day_data = sync_single_k_type('day', all_stocks, global_login_success)
        week_success, week_data = sync_single_k_type('week', all_stocks, global_login_success)
        # month_success, month_data = sync_single_k_type('month', all_stocks, global_login_success)

        # 任务总结
        overall_end = time.time()
        overall_elapsed = int(overall_end - overall_start)
        logger.info("\n" + "=" * 50)
        logger.info("===== 同步任务全部完成 =====")
        # logger.info(f"日K同步：成功{day_success}只股票，插入{day_data}条数据")
        logger.info(f"周K同步：成功{week_success}只股票，插入{week_data}条数据（含多维度分析）")
        # logger.info(f"月K同步：成功{month_success}只股票，插入{month_data}条数据")
        logger.info(f"总耗时：{overall_elapsed // 3600}小时{(overall_elapsed % 3600) // 60}分钟{overall_elapsed % 60}秒")
        logger.info("=" * 50)

    finally:
        if global_login_success:
            bs.logout()
            logger.info("\n=== Baostock全局登出成功 ===")


if __name__ == "__main__":
    main()